package cnooc.mqtt;

import cn.hutool.core.util.StrUtil;
import cnooc.common.config.InfluxDbComUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class MessageArrivedCallback implements MqttCallback {
    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("连接丢失: " + cause.getMessage());
    }
    @Resource
    private InfluxDbComUtil influxDbComUtil;
    @Resource
    RedisTemplate<String, Object> redisTemplate;
    private static final String tableNameKey = "ems_equipment:{}:tableName";
    private static final String locationKey = "ems_equipment:{}:locations";
    @Override
    public void messageArrived(String topic, MqttMessage message) {
//        System.out.println("接收到消息: 主题 = " + topic + "，内容 = " + new String(message.getPayload()));
        JSONObject jsonObject = JSON.parseObject(message.getPayload());
        Object devs = jsonObject.get("devs");
        List<Map<String, Object>> list = (List<Map<String, Object>>) devs;
        for (Map<String, Object> map : list) {
            String devId = map.get("dev").toString();
            String currentLocationKey = StrUtil.format(locationKey, devId);
            String currentTableNameKey = StrUtil.format(tableNameKey, devId);
            if (!redisTemplate.hasKey(currentLocationKey) || !redisTemplate.hasKey(currentTableNameKey)) continue;
            String tableName = redisTemplate.opsForValue().get(currentTableNameKey).toString();
            Map<String, Object> fields = new HashMap<>();
            List<Map<String, String>> values= (List<Map<String, String>>) map.get("d");
            for (Map<String, String> value : values) {
                String key = value.get("m");
                String v = value.get("v");
                fields.put(key, Float.parseFloat(v));
            }
            if (!fields.isEmpty()) {
                Map tags = new HashMap();
                tags.put("dev_id", devId);
                influxDbComUtil.insert(tableName, fields, tags);
            }
        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 不需要处理（订阅者不发送消息）
    }
}
